package com.atguigu.bigdata.edu.realtime.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.bigdata.edu.realtime.app.BaseAppV1;
import com.atguigu.bigdata.edu.realtime.bean.AnswerDetailBean;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.AtguiguUtil;
import com.atguigu.bigdata.edu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.Duration;

public class Dws_16_DwsTestQuestionWindow extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_16_DwsTestQuestionWindow().init(4521, 2, "Dws_16_DwsTestQuestionWindow", Constant.DWD_EXAM_QUESTION_DETAIL);
    }
    //TODO 这个需求在于需要做两次keyby 这里使用一次tuple做第一次keyby之后 再用courseId做keyby聚合开窗写出
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        SingleOutputStreamOperator<AnswerDetailBean> pojo = parseToPOJO(stream);

        SingleOutputStreamOperator<AnswerDetailBean> answerDetailBeanSingleOutputStreamOperator = windowAndAgg(pojo);

        answerDetailBeanSingleOutputStreamOperator.addSink(FlinkSinkUtil.getClickHouseSink("dws_test_question_window",AnswerDetailBean.class));


    }

    private SingleOutputStreamOperator<AnswerDetailBean> windowAndAgg(SingleOutputStreamOperator<AnswerDetailBean> pojo) {
        SingleOutputStreamOperator<AnswerDetailBean> reduce = pojo.assignTimestampsAndWatermarks(WatermarkStrategy.<AnswerDetailBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withIdleness(Duration.ofSeconds(3))
                        .withTimestampAssigner((o, ts) -> o.getTs()))
                .keyBy(new KeySelector<AnswerDetailBean, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> getKey(AnswerDetailBean value) throws Exception {
                        return Tuple2.of(value.getQuestionId(), value.getUid());
                    }
                }).reduce(new ReduceFunction<AnswerDetailBean>() {
                    @Override
                    public AnswerDetailBean reduce(AnswerDetailBean value1, AnswerDetailBean value2) throws Exception {
                        value1.setCorrectCt(value1.getCorrectCt() + value2.getCorrectCt());
                        value1.setAnswerCt(value1.getAnswerCt() + value2.getAnswerCt());
                        if (!(value1.getCorrectUserCt() == 1)) {
                            value1.setCorrectUserCt(value2.getCorrectUserCt());
                        }
                        return value1;
                    }
                });
        return reduce.assignTimestampsAndWatermarks(WatermarkStrategy.<AnswerDetailBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withIdleness(Duration.ofSeconds(3))
                        .withTimestampAssigner((o, ts) -> o.getTs()))
                .keyBy(AnswerDetailBean::getQuestionId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<AnswerDetailBean>() {
                    @Override
                    public AnswerDetailBean reduce(AnswerDetailBean value1, AnswerDetailBean value2) throws Exception {
                        value1.setCorrectCt(value1.getCorrectCt() + value2.getCorrectCt());
                        value1.setAnswerCt(value1.getAnswerCt() + value2.getAnswerCt());

                        value1.setCorrectUserCt(value1.getCorrectUserCt() + value2.getCorrectUserCt());
                        value1.setAnswerUserCt(value1.getAnswerUserCt() + value2.getAnswerUserCt());
                        value1.setCorrectUserRate(BigDecimal.valueOf(value1.getCorrectUserCt()).divide(new BigDecimal(value1.getAnswerUserCt()), 2, RoundingMode.UP));
                        return value1;
                    }
                }, new ProcessWindowFunction<AnswerDetailBean, AnswerDetailBean, String, TimeWindow>() {

                    @Override
                    public void process(String s, ProcessWindowFunction<AnswerDetailBean, AnswerDetailBean, String, TimeWindow>.Context context, Iterable<AnswerDetailBean> elements, Collector<AnswerDetailBean> out) throws Exception {
                        AnswerDetailBean next = elements.iterator().next();
                        next.setCorrectRate(BigDecimal.valueOf(next.getCorrectCt()).divide(new BigDecimal(next.getAnswerCt()), 2, RoundingMode.UP));
                        next.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                        next.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));
                        next.setTs(context.currentWatermark());
                        out.collect(next);
                    }
                });

    }

    private SingleOutputStreamOperator<AnswerDetailBean> parseToPOJO(DataStreamSource<String> stream) {
        return stream.map(new MapFunction<String, AnswerDetailBean>() {
            @Override
            public AnswerDetailBean map(String value) throws Exception {
                JSONObject obj = JSONObject.parseObject(value);
                String exam_id = obj.getString("exam_id");
                String question_id = obj.getString("question_id");
                String question_txt = obj.getString("question_txt");
                String user_id = obj.getString("user_id");
                Long is_correct = obj.getLong("is_correct");
                Long ts = obj.getLong("create_time");
                return new AnswerDetailBean("", "", user_id, question_id, question_txt, is_correct, 1L, BigDecimal.valueOf(0), is_correct, 1L, BigDecimal.valueOf(0), ts);
            }
        });

    }
}
